/*
 *  Arrow
 *
 *  Created by David Giovannini on 1/5/07.
 *  Copyright 2007 Software by Jove. All rights reserved.
 *
 */
 
#include "SBJ/Arrow/Dispatcher.h"
#include "SBJ/Arrow/Subscription.h"
#include "SBJ/Arrow/MsgProxy.h"

namespace SBJ
{

namespace Arrow
{

Dispatcher::Dispatch::Dispatch( Message* msg )
: m_blocker( 0 )
, m_msg( msg )
, m_subscriber( 0 )
{
}

Dispatcher::Dispatch::Dispatch( Message* msg, Blocker& blocker )
: m_msg( msg )
, m_blocker( blocker )
, m_subscriber( 0 )
{
}

Dispatcher::Dispatch::Dispatch( Message* msg, Blocker& blocker, void* subscriber )
: m_msg( msg )
, m_blocker( blocker )
, m_subscriber( subscriber )
{
}

void
Dispatcher::Dispatch::operator () ( void )
{
	if ( m_subscriber == 0 )
	{
		if ( m_blocker.hasSubstance() )
		{
			m_blocker.wait();
			m_msg->getTopic().getMsgProxy().endInvocations( *m_msg );
		}
		else
		{
			m_msg->getTopic().getMsgProxy().execute( *m_msg );
		}
		delete m_msg;
	}
	else
	{
		m_msg->getTopic().getMsgProxy().invoke( *m_msg, m_subscriber );
		m_blocker.finish();
	}
}

Dispatcher::Dispatcher( int numDispatchingThreads )
: m_threadPool( numDispatchingThreads )
, m_workByTopicIter( m_byTopic.end() )
, m_numThreads( numDispatchingThreads )
{
}
	
Dispatcher::~Dispatcher( void )
{
}

bool
Dispatcher::subscribe( const Subscription& subscription )
{
	const Topic& topic = subscription.getTopic();
	if ( subscribe( subscription.getSubscriber(), subscription.getTopic() ) )
	{
		m_stats.accumeSubscribe();
		topic.addUse();
		return true;
	}
	return false;
}

bool
Dispatcher::unsubscribe( const Subscription& subscription )
{
	const Topic& topic = subscription.getTopic();
	if ( unsubscribe( subscription.getSubscriber(), subscription.getTopic() ) )
	{
		m_stats.accumeUnsubscribe();
		topic.removeUse();
		return true;
	}
	return false;
}

bool
Dispatcher::subscribe( void* subscriber, const Topic& topic )
{
	boost::mutex::scoped_lock lock( m_subscriberGuard );

	// Make certain we know about the topic
	KeyedSubscribers::iterator subscriberListIter =
	m_byTopic.insert( std::make_pair( topic, Subscribers() ) ).first;

	// If we want to modify the set we are currently iterating over
	// delay this work
	if ( m_workByTopicIter == subscriberListIter )
	{
		Work work = { task_add, subscriber, topic };
		m_delayedTasks.push_back( work );
		return true;
	}

	// Get the set of subscribers for this topic
	Subscribers& subscribers = subscriberListIter->second;

	// Add the subscriber to this list
	bool newSubscription = subscribers.insert( subscriber ).second;

	// If not previously subscribed
	if ( newSubscription == true )
	{
		KeyedTopics::iterator topicListIter =
			m_bySubscriber.insert( std::make_pair( subscriber, Topics() ) ).first;
		    
		// Append the subscription
		topicListIter->second.insert( topic );
	}

	return true;
}
	
bool
Dispatcher::unsubscribe( void* subscriber, const Topic& topic )
{
	boost::mutex::scoped_lock lock( m_subscriberGuard );

	bool success = false;
	// Get our subscriber list for this topic
	KeyedSubscribers::iterator subscriberListIter = m_byTopic.find( topic );
	if ( subscriberListIter != m_byTopic.end() )
	{
		// If we want to modify the set we are currently iterating over
		// delay this work
		if ( m_workByTopicIter == subscriberListIter )
		{
			Work work = { task_remove, subscriber, topic };
			m_delayedTasks.push_back( work );
			return true;
		}

		// Get our subscriber list for this topic
		Subscribers& subscribers = subscriberListIter->second;

		Subscribers::iterator findSubscriberIter = subscribers.find( subscriber );
		if ( findSubscriberIter != subscribers.end() )
		{
			if ( subscribers.size() == 0 )
			{
				m_byTopic.erase( subscriberListIter );
			}
			else
			{
				subscribers.erase( findSubscriberIter );
			}
			success = true;
		}
	}

	if ( success == true )
	{
		// Get our topic list for this subscriber
		KeyedTopics::iterator topicListIter = m_bySubscriber.find( subscriber );
		if ( topicListIter != m_bySubscriber.end() )
		{
			// Get our subscriber list for this topic
			Topics& topics = topicListIter->second;

			Topics::iterator findTopicIter = topics.find( topic );
			if ( findTopicIter != topics.end() )
			{
				if ( topics.size() == 0 )
				{
					m_bySubscriber.erase( topicListIter );
				}
				else
				{
					topics.erase( findTopicIter );
				}
			}
		}
	}
	return success;
}

void
Dispatcher::performTasks( void )
{
	std::list< Work >::iterator iter( m_delayedTasks.begin() );
	while ( iter != m_delayedTasks.end() )
	{
		switch ( iter->task )
		{
			case task_add:
				subscribe( iter->subscriber, iter->topic );
				break;
			case task_remove:
				unsubscribe( iter->subscriber, iter->topic );
				break;
			default:
				break;
		}
		++iter;
	}
}
	
void
Dispatcher::onMessage( Message& msg )
{
	const Topic& topic = msg.getTopic();
	const MsgProxy& proxy = topic.getMsgProxy();

	size_t threadedThreshold = proxy.getThreadedThreshold( msg );
	boost::tribool blocks = proxy.doesBlockOnDispatch( msg );

	if ( boost::indeterminate( blocks ) )
	{
		m_stats.accumeDead();
		msg.getTopic().getMsgProxy().dropped( msg, Message::dropped_UncaughtException );
	}

	// If message gets executed
	if ( topic.isEmpty() == true )
	{
		m_stats.accumeExecute();
		if ( threadedThreshold <= 1 && blocks == false && m_numThreads > 0 )
		{
			m_stats.accumeThreaded( 1 );
            Dispatch exec( proxy.duplicate( msg ) );
			m_threadPool.addTask( exec );
		}
		else
		{
			m_stats.accumeBlocking( 1 );
			proxy.execute( msg );
		}
		return;
	}

	/// Find the list of subscribers for the Topic being dispatched
	KeyedSubscribers::iterator subscriberListIter = m_byTopic.find( topic );

	// If message has no recipients
	if ( subscriberListIter == m_byTopic.end() )
	{
		m_stats.accumeDead();
		proxy.dropped( msg, Message::dropped_NoSubscribers );
		return;
	}
	// If message visits recipients
	else
	{
		m_stats.accumeVisitBegin();
		{
			boost::mutex::scoped_lock lock( m_subscriberGuard );
			m_workByTopicIter = subscriberListIter;
		}

		Subscribers& subscribers = subscriberListIter->second;
		Subscribers::iterator findSubscriberIter = subscribers.begin();
		Subscribers::iterator endSubscriberIter = subscribers.end();
		int subscriberCount = static_cast< int >( subscribers.size() );

		// Determine if we do not want want threaded execution
		if ( subscriberCount < static_cast< int >( threadedThreshold ) || ( subscriberCount == 1 && blocks ) || m_numThreads == 0 )
		{
			m_stats.accumeBlocking( subscriberCount );
			boost::tribool begin = proxy.beginInvocations( msg, subscriberCount );
			if ( begin )
			{
				m_stats.accumeVisitations( subscriberCount );
				while ( findSubscriberIter != endSubscriberIter )
				{
					proxy.invoke( msg, *findSubscriberIter );
					findSubscriberIter++;
				}
				proxy.endInvocations( msg );
			}
			else if ( !begin )
			{
				m_stats.accumeCancel();
				m_stats.accumeBlocking( 1 );
				proxy.dropped( msg, Message::dropped_VisitationCanceled );
			}
			else
			{
				m_stats.accumeDead();
				proxy.dropped( msg, Message::dropped_UncaughtException );
			}
		}
		else
		{
			Message* sharedMsg = proxy.duplicate( msg );
			boost::tribool begin = proxy.beginInvocations( *sharedMsg, subscriberCount );

			if ( begin )
			{
				Blocker blocker( subscriberCount );

				// For each subscriber add the dispatch function
				m_stats.accumeThreaded( subscriberCount );
				m_stats.accumeVisitations( subscriberCount );
				while ( findSubscriberIter != endSubscriberIter )
				{
                    Dispatch invoke( sharedMsg, blocker, *findSubscriberIter );
					m_threadPool.addTask( invoke );
					findSubscriberIter++;
				}

				// Wait until all the threaded tasks are complete.
				if ( blocks == true )
				{
					Dispatch dispatch( sharedMsg, blocker );
					dispatch();
				}
				// queue up a task that will complete the message
				else
				{
                    Dispatch end( sharedMsg, blocker );
					m_threadPool.addTask( end );
				}
			}
			else if ( !begin )
			{
				m_stats.accumeCancel();
				m_stats.accumeBlocking( 1 );
				proxy.dropped( msg, Message::dropped_VisitationCanceled );
			}
			else
			{
				m_stats.accumeDead();
				proxy.dropped( msg, Message::dropped_UncaughtException );
			}
		}

		{
			boost::mutex::scoped_lock lock( m_subscriberGuard );
			m_workByTopicIter = m_byTopic.end();
			performTasks();
		}
	}	// if one or more Subscribers have interest in this Topic/Topic
}

}

}
